//package com.sdgs.paas.sdk.core.mq;
//
//import java.io.UnsupportedEncodingException;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//import java.util.Map.Entry;
//import java.util.concurrent.ConcurrentHashMap;
//import java.util.concurrent.atomic.AtomicInteger;
//
//import org.springframework.context.ApplicationContext;
//
//import com.ctg.mq.api.IMQAckHandler;
//import com.ctg.mq.api.bean.MQResult;
//import com.ctg.mq.api.exception.MQException;
//import com.ctg.mq.api.listener.ConsumeListener;
//import com.sdgs.paas.sdk.core.log.PaasLogger;
//
///**
// * ctgmq消息监听器
// *
// * @author wjj
// *
// */
//
//public class CtgConsumeListener implements ConsumeListener {
//	private ApplicationContext ctx;
//	private Boolean start = false;
//	private AtomicInteger consumeMsgCounter = new AtomicInteger(0);
//	private Map<List<MQResult>, IMQAckHandler> msgCache = new HashMap<>();
//	private Class<?> strategyClass;
//	private ICtgMqConsumStrategy ctgMqConsumStrategy;
//
//	@Override
//	public void onMessage(List<MQResult> mqResultList, IMQAckHandler ackHandler) {
//		if (null == mqResultList || mqResultList.isEmpty()) {
//			PaasLogger.warn("没有消息可供消费");
//			return;
//		}
//		// 锁住start对象，使onMessage和startDealmsg在同一时间只有一个方法在运行
//		synchronized (start) {
//			if (start) {
//				dealmsg(mqResultList, ackHandler);
//			} else {
//				msgCache.put(mqResultList, ackHandler);
//				return;
//			}
//		}
//
//	}
//
//	public void startDealmsg() {
//		// 锁住start对象，使onMessage和startDealmsg在同一时间只有一个方法在运行
//		synchronized (start) {
//			if (!start) {
//				ctgMqConsumStrategy = (ICtgMqConsumStrategy) ctx.getBean(strategyClass);
//
//				for (Entry<List<MQResult>, IMQAckHandler> entry : msgCache.entrySet()) {
//					dealmsg(entry.getKey(), entry.getValue());
//				}
//
//				start = true;
//			}
//		}
//	}
//
//	private void dealmsg(List<MQResult> mqResultList, IMQAckHandler ackHandler) {
//		// 拉取到消息后，一条条来消费，并ack
//		for (MQResult mqResult : mqResultList) {
//			consumeMsgCounter.incrementAndGet();
//			// PaasLogger.info(Thread.currentThread().getName()
//			// +
//			// "消费第" + consumeMsgCounter.get() + "条消息：["
//			// + mqResult.toString() + "]");
//
//			/**
//			 * 进行消费， 业务代码
//			 */
//			try {
//				long start = System.currentTimeMillis();
//				PaasLogger.info("{}消费第{}条消息：[{}]开始!", Thread.currentThread().getName(), consumeMsgCounter.get(),
//						mqResult.toString());
//				ctgMqConsumStrategy.dealMsg(new String(mqResult.getMessage().getBody(), "utf-8"),
//						mqResult.getMessage().getKey(), mqResult.getMessage().getTag());
//				PaasLogger.info("{}消费第{}条消息：[{}]结束,处理耗时 {} 毫秒!", Thread.currentThread().getName(),
//						consumeMsgCounter.get(), mqResult.toString(), System.currentTimeMillis() - start);
//
//			} catch (UnsupportedEncodingException e1) {
//				PaasLogger.error(e1, "消息处理时出现错误:UnsupportedEncodingException");
//			} catch (Exception e1) {
//				PaasLogger.error(e1, "消息处理时出现错误：");
//			}
//
//			// 消费完后，进行ack
//			try {
//				ackHandler.ackMessageSuccess(mqResult);
//			} catch (MQException e) {
//				PaasLogger.error(e,
//						e.getExpDesc());/**
//										 * ack异常的， 需要记录错误日志或根据错误码做进一步处理
//										 */
//				// .................................
//			}
//		}
//	}
//
//	public Boolean getStart() {
//		return start;
//	}
//
//	public void setStart(Boolean start) {
//		this.start = start;
//	}
//
//	public AtomicInteger getConsumeMsgCounter() {
//		return consumeMsgCounter;
//	}
//
//	public void setConsumeMsgCounter(AtomicInteger consumeMsgCounter) {
//		this.consumeMsgCounter = consumeMsgCounter;
//	}
//
//	public Map<List<MQResult>, IMQAckHandler> getMsgCache() {
//		return msgCache;
//	}
//
//	public void setMsgCache(Map<List<MQResult>, IMQAckHandler> msgCache) {
//		this.msgCache = msgCache;
//	}
//
//	public ICtgMqConsumStrategy getCtgMqConsumStrategy() {
//		return ctgMqConsumStrategy;
//	}
//
//	public void setCtgMqConsumStrategy(ICtgMqConsumStrategy ctgMqConsumStrategy) {
//		this.ctgMqConsumStrategy = ctgMqConsumStrategy;
//	}
//
//	public Class<?> getStrategyClass() {
//		return strategyClass;
//	}
//
//	public void setStrategyClass(Class<?> strategyClass) {
//		this.strategyClass = strategyClass;
//	}
//
//	public ApplicationContext getCtx() {
//		return ctx;
//	}
//
//	public void setCtx(ApplicationContext ctx) {
//		this.ctx = ctx;
//	}
//
//}